package com.atguigu.medical.realtime.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.medical.realtime.util.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lzc
 * @Date 2023/5/5 11:19
 */
public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T, T> implements DimFunction<T> {
    
    private ThreadPoolExecutor threadPool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
    }
    
    // 异步调用函数:
    // input: 需要异步处理的元素
    // resultFuture: 用来输出最终异步处理后元素
    @Override
    public void asyncInvoke(T bean,
                            ResultFuture<T> resultFuture) throws Exception {
        // 因为没有异步客户端: 所以需要没来一个元素,就使用线程对这元素进行处理
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                // 线程内执行的代码
                // 1. 获取 redis 客户端和 phoenix 客户端
                Jedis redisClient = RedisUtil.getRedisClient();
                Connection conn = DruidDSUtil.getConnection();
                JSONObject dim = DimUtil.readDim(redisClient, conn, getTable(), getId(bean));
                addDim(bean, dim);  // 执行结束之后, bean 中的维度就补充完毕
    
                // 连接对象,用完之后,一定要关闭
                redisClient.close();
                JdbcUtil.closeConnection(conn);
    
                // 把维度补充完毕的 bean 输出
                resultFuture.complete(Collections.singletonList(bean));
            }
        });
    }
    
    // 如果异步超时,会执行这个方法
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        super.timeout(input, resultFuture);
        // 根据自己情况进行处理
        throw new RuntimeException("异步超时:一般是其他原因导致的一部超时, 请检查: \n" +
                                       "1.检查集群是否都 ok: hdfs redis hbase kafka. \n" +
                                       "2.检查下 redis 的配置 bind 0.0.0.0 \n" +
                                       "3.检查用到的维度表是否都在,并且每张表都数据 \n" +
                                       "4. 找我" +
                                       "");
    }
}
